package com.zliuyno.rpcdemo.transport;


import com.zliuyno.rpcdemo.rpc.Protocol.MyContent;
import com.zliuyno.rpcdemo.rpc.Protocol.MyHeader;
import com.zliuyno.rpcdemo.rpc.ResponseMappingCallback;
import com.zliuyno.rpcdemo.util.SerDerUtil;
import io.netty.bootstrap.Bootstrap;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.PooledByteBufAllocator;
import io.netty.buffer.Unpooled;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.http.*;

import java.io.*;
import java.net.*;
import java.util.Random;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;

public class ClientFactory {

    int poolSize = 1;

    NioEventLoopGroup group;

    Random random = new Random();

    private ClientFactory() {
    }

    private static final ClientFactory factory;

    static {
        factory = new ClientFactory();
    }

    public static ClientFactory getFactory() {
        return factory;
    }


    public static CompletableFuture<Object> transport(MyContent content) {
        //content就是货物,现在可以用自定义的rpc传输协议(有状态)，也可以用http协议作为载体
        //我们先用http协议作为载体，那样未来就可以让provider是一个tomcat jetty  这种基于htto协议的容器
        //有无状态来自于你使用什么协议, hhtp无状态   没请求对应一个连接
        //dubbo是一个rpc框架 netty是一个io框架
        //dubbo中传输协议上，可以是自定义的rpc传输协议 也可以是http协议

        CompletableFuture<Object> res = new CompletableFuture<>();
//        String type="rpc";
        String type = "http";

        if ("rpc".equals(type)) {
            byte[] msgBody = SerDerUtil.ser(content);
            MyHeader header = MyHeader.createHeader(msgBody);
            byte[] msgHeader = SerDerUtil.ser(header);
//            System.out.println("main:: :" + msgHeader.length);

            //3.连接池 ：取得连接
            ClientFactory factory = ClientFactory.getFactory();
            NioSocketChannel clientChannel = factory.getClient(new InetSocketAddress("localhost", 9090));

            //获取连接的过程   开始-创建   连接中-直接取
            //4.发送--> 走 IO out --> 走Netty（event 驱动）
            ByteBuf byteBuf = PooledByteBufAllocator.DEFAULT.directBuffer(msgHeader.length + msgBody.length);
            long requestID = header.getRequestID();

            ResponseMappingCallback.addCallBack(requestID, res);

            byteBuf.writeBytes(msgHeader);
            byteBuf.writeBytes(msgBody);

            ChannelFuture channelFuture = clientChannel.writeAndFlush(byteBuf);
        } else {
            //使用http协议为载体
            //1.用URL 现成的工具(包含了http的编解码,发送,socket,连接)
//            urlTS(content, res);
            //2.自己操心：on netty + 已经提供的http相关的编解码
            nettyTS(content, res);

        }
        return res;
    }

    private static void nettyTS(MyContent content, CompletableFuture<Object> res) {


        //1.通过netty建立io 建立连接
        NioEventLoopGroup group = new NioEventLoopGroup(1); //应该定义到外面
        Bootstrap bs = new Bootstrap();
        Bootstrap client = bs.group(group)
                .channel(NioSocketChannel.class)
                .handler(new ChannelInitializer<NioSocketChannel>() {
                    @Override
                    protected void initChannel(NioSocketChannel nioSocketChannel) throws Exception {
                        ChannelPipeline p = nioSocketChannel.pipeline();
                        p.addLast(new HttpClientCodec())
                                .addLast(new HttpObjectAggregator(1024 * 512))
                                .addLast(new ChannelInboundHandlerAdapter() {
                                    @Override
                                    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
                                        //3.接收  预埋的回调，根据netty对socket io事件的响应
                                        //客户端的msg是啥：完整的http的Response
                                        FullHttpResponse response = (FullHttpResponse) msg;
                                        System.out.println(response.toString());

                                        ByteBuf resContent = response.content();
                                        byte[] data = new byte[resContent.readableBytes()];
                                        resContent.readBytes(data);
                                        ObjectInputStream oin = new ObjectInputStream(new ByteArrayInputStream(data));
                                        MyContent o = (MyContent) oin.readObject();

                                        res.complete(o.getRes());
                                    }
                                });
                    }
                });
        try {
            //2.发送
            ChannelFuture channelFuture = client.connect("localhost", 9090).sync();
            Channel clientChannel = channelFuture.channel();
            byte[] data = SerDerUtil.ser(content);
            DefaultFullHttpRequest request = new DefaultFullHttpRequest
                    (HttpVersion.HTTP_1_0, HttpMethod.POST, "/", Unpooled.copiedBuffer(data));

            request.headers().set(HttpHeaderNames.CONTENT_LENGTH,data.length);
            clientChannel.writeAndFlush(request).sync();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }




    }

    private static void urlTS(MyContent content, CompletableFuture<Object> res) {
        //这种方式是每请求占用一个连接，因为使用的是http协议
        Object obj = null;
        try {
            URL url = new URL("http://localhost:9090/");
            HttpURLConnection hc = (HttpURLConnection) url.openConnection();
            hc.setRequestMethod("POST");
            hc.setDoOutput(true);
            hc.setDoInput(true);

            OutputStream out = hc.getOutputStream();
            ObjectOutputStream oout = new ObjectOutputStream(out);
            oout.writeObject(content);

            if (hc.getResponseCode() == 200) {//此环节才真正发出去
                InputStream in = hc.getInputStream();
                ObjectInputStream oin = new ObjectInputStream(in);
                MyContent myContent = (MyContent) oin.readObject();

                obj = myContent.getRes();
            }

        } catch (Exception e) {
            e.printStackTrace();
        }
        res.complete(obj);
    }

    //一个consumer可以连接多个provider,每一个provider都有自己的pool   K,V
    ConcurrentHashMap<InetSocketAddress, ClientPool> outboxs = new ConcurrentHashMap<>();


    public synchronized NioSocketChannel getClient(InetSocketAddress address) {
        //TODO: 并发情况下一定要谨慎

        ClientPool clientPool = outboxs.get(address);
        if (clientPool == null) {
            synchronized (outboxs) {
                if (clientPool == null) {
                    outboxs.putIfAbsent(address, new ClientPool(poolSize));
                    clientPool = outboxs.get(address);
                }
            }
        }
        int i = random.nextInt(poolSize);
        if (clientPool.clients[i] != null && clientPool.clients[i].isActive()) {
            return clientPool.clients[i];
        } else {
            synchronized (clientPool.lock[i]) {
                if (clientPool.clients[i] == null || !clientPool.clients[i].isActive()) {
                    clientPool.clients[i] = create(address);
                }
            }
            return clientPool.clients[i];
        }
    }

    private NioSocketChannel create(InetSocketAddress address) {
        //基于netty的客户端创建方式
        group = new NioEventLoopGroup(1);
        Bootstrap bs = new Bootstrap();
        ChannelFuture connect = bs.group(group)
                .channel(NioSocketChannel.class)
                .handler(new ChannelInitializer<NioSocketChannel>() {
                    @Override
                    protected void initChannel(NioSocketChannel nioSocketChannel) throws Exception {
                        ChannelPipeline p = nioSocketChannel.pipeline();
                        p.addLast(new ServerDecode());
                        p.addLast(new ClientResponses());    //解决给谁的 谁去处理这个返回
                    }
                }).connect(address);
        Channel client = null;
        try {
            client = connect.sync().channel();
            return (NioSocketChannel) client;
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        return null;
    }
}